[FLINK-38461] Introduce SinkUpsertMaterializerV2 #27070

rkhachatryan · 2025-09-30T15:22:29Z

This PR adds a new version of SinkUpsertMaterializer based on Sequenced MultiSet state introduced in #27071.

The existing operator is not changed.

Test coverage is provided by parameterized the tests extended in #27068.

flinkbot · 2025-09-30T15:27:34Z

CI report:

6a58e7f Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

This allows to use RowTypeUtils both in planner and in runtime

1996fanrui

Thanks @rkhachatryan for the pr!

Overall LGTM, I left some comments, please take a look when you are available, thanks

1996fanrui · 2025-10-15T15:27:16Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java

+            TABLE_EXEC_SINK_UPSERT_MATERIALIZE_STRATEGY =
+                    key("table.exec.sink.upsert-materialize.strategy")
+                            .enumType(SinkUpsertMaterializeStrategy.class)
+                            .noDefaultValue()


The default is LEGACY, why noDefaultValue() here?

1996fanrui · 2025-10-15T15:29:05Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java

+                                                    "When using strategy=ADAPTIVE, defines the number of entries per key when the implementation is changed from MAP to VALUE. If not specified, Flink uses state-backend specific defaults.")
+                                            .linebreak()
+                                            .text("The option takes effect during job (re)starting")
+                                            .linebreak()
+                                            .build());
+
+    @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING)
+    public static final ConfigOption<Long>
+            TABLE_EXEC_SINK_UPSERT_MATERIALIZE_ADAPTIVE_THRESHOLD_HIGH =
+                    key("table.exec.sink.upsert-materialize.adaptive.threshold.high")
+                            .longType()
+                            .noDefaultValue()
+                            .withDescription(
+                                    Description.builder()
+                                            .text(
+                                                    "When using strategy=ADAPTIVE, defines the number of entries per key when the implementation is changed from VALUE to MAP. If not specified, Flink uses state-backend specific defaults.")


Would you mind mentioning the specific default value for each state backend?

1996fanrui · 2025-10-15T15:32:57Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java

+                                            .text(
+                                                    "When using strategy=ADAPTIVE, defines the number of entries per key when the implementation is changed from VALUE to MAP. If not specified, Flink uses state-backend specific defaults.")
+                                            .linebreak()
+                                            .text("The option takes effect during job (re)starting")


I do not understand why mentioning "The option takes effect during job (re)starting". As I understand, almost all of config options take effect after job (re)starting.

Do you wanna point out that these 2 config options cannot be changed via dynamic conf?

1996fanrui · 2025-10-15T15:36:35Z

...ble/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ProjectedRowData.java

                + Arrays.toString(indexMapping)
                + ", mutableRow="
                + row
                + '}';


nit: how about adding isNullAtNonProjected into toString() as well?

1996fanrui · 2025-10-15T15:40:58Z

...rg/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalOverAggregateRule.scala

 import org.apache.flink.table.planner.plan.utils.{AggregateUtil, OverAggregateUtil, SortUtil}
 import org.apache.flink.table.planner.plan.utils.PythonUtil.isPythonAggregate
 import org.apache.flink.table.planner.utils.ShortcutUtils
+import org.apache.flink.table.typeutils.RowTypeUtils


This file does not change any code, why import is needed? by mistake or do I miss something?

1996fanrui · 2025-10-15T15:46:14Z

...ain/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalSink.scala

      upsertMaterialize,
+      // persist upsertMaterialize strategy separately in the compiled plan to make it immutable;
+      // later on, it can't be obtained from the node config because it is merged with the new environment
+      config.getOptional(TABLE_EXEC_SINK_UPSERT_MATERIALIZE_STRATEGY).orElse(null),


Why it can be null?

1996fanrui · 2025-10-15T16:02:31Z

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java

+                        config.getOptional(
+                                        TABLE_EXEC_SINK_UPSERT_MATERIALIZE_ADAPTIVE_THRESHOLD_HIGH)
+                                .orElse(null),
+                        config.getOptional(
+                                        TABLE_EXEC_SINK_UPSERT_MATERIALIZE_ADAPTIVE_THRESHOLD_LOW)
+                                .orElse(null),


Suggested change

config.getOptional(

TABLE_EXEC_SINK_UPSERT_MATERIALIZE_ADAPTIVE_THRESHOLD_HIGH)

.orElse(null),

config.getOptional(

TABLE_EXEC_SINK_UPSERT_MATERIALIZE_ADAPTIVE_THRESHOLD_LOW)

.orElse(null),

config.get(TABLE_EXEC_SINK_UPSERT_MATERIALIZE_ADAPTIVE_THRESHOLD_HIGH),

config.get(TABLE_EXEC_SINK_UPSERT_MATERIALIZE_ADAPTIVE_THRESHOLD_LOW),

It they are not set, and no default value, get will return null directly.

1996fanrui · 2025-10-15T16:11:44Z

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java

+                        : HashCodeGenerator.generateRowHash(
+                                new CodeGeneratorContext(config, classLoader),
+                                RowTypeUtils.projectRowType(physicalRowType, inputUpsertKey),
+                                "",


how about setting a name here?

1996fanrui · 2025-10-15T16:20:33Z

.../src/test/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializerTest.java

+                        TTL_CONFIG, RowType.of(types), EQUALISER, UPSERT_KEY_EQUALISER, upsertKey);
+            case MAP:
+                return createV2(
+                        types,
+                        upsertKey,
+                        SequencedMultiSetStateConfig.forMap(
+                                PROCESSING_TIME, StateTtlConfig.DISABLED));
+            case VALUE:
+                return createV2(
+                        types,
+                        upsertKey,
+                        SequencedMultiSetStateConfig.forValue(
+                                PROCESSING_TIME, StateTtlConfig.DISABLED));
+            case ADAPTIVE:
+                return createV2(
+                        types,
+                        upsertKey,
+                        SequencedMultiSetStateConfig.adaptive(
+                                PROCESSING_TIME, 10L, 5L, StateTtlConfig.DISABLED));


how about extracting a method to create StateTtlConfig via isTtlSupported() method?

It is helpful for keep the code consistency, only change isTtlSupported() in the future while supporting TTL.

1996fanrui · 2025-10-16T09:42:56Z

...me/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializerV2.java

+                                        ? ProjectedRowData.from(inputUpsertKey)
+                                                .withNullAtNonProjected(true)
+                                                .replaceRow(row)
+                                        : row,


I have 2 comments about this part since it is a hot code path:

withNullAtNonProjected will create a new ProjectedRowData object, so it created 2 ProjectedRowData objects for each row.

ProjectedRowData is created on open in SinkUpsertMaterializer[1], and then it is reused by all rows. So how about create ProjectedRowData only once as well here.

[1]

flink/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializer.java

Line 123 in e44d638

upsertKeyProjectedRow2 = ProjectedRowData.from(inputUpsertKey);

This was referenced Sep 30, 2025

[FLINK-38460][tests] Improve test coverage for SinkUpsertMaterializer #27068

Merged

[FLINK-38464] Introduce OrderedMultiSetState #27071

Merged

rkhachatryan added 3 commits October 15, 2025 08:57

[hotfix] Move RowTypeUtils to flink-table-common

d4c53a0

This allows to use RowTypeUtils both in planner and in runtime

[hotfix] Introduce ProjectedRowData.isNullAtNonProjected

59fd50d

[FLINK-38461] Introduce SinkUpsertMaterializerV2

42fd98e

rkhachatryan force-pushed the f38461 branch from 1a043c8 to 42fd98e Compare October 15, 2025 07:42

rkhachatryan changed the title ~~[DRAFT][FLINK-38461] Introduce SinkUpsertMaterializerV2~~ [FLINK-38461] Introduce SinkUpsertMaterializerV2 Oct 15, 2025

rkhachatryan marked this pull request as ready for review October 15, 2025 07:47

1996fanrui self-assigned this Oct 15, 2025

rkhachatryan requested a review from 1996fanrui October 15, 2025 12:07

rkhachatryan added 2 commits October 15, 2025 12:30

fixup: checkstyle

9b99ec7

fixup: adjust SUM v2 creation

6a58e7f

1996fanrui reviewed Oct 16, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-38461] Introduce SinkUpsertMaterializerV2 #27070

[FLINK-38461] Introduce SinkUpsertMaterializerV2 #27070

rkhachatryan commented Sep 30, 2025 •

edited

Loading

Uh oh!

flinkbot commented Sep 30, 2025 •

edited

Loading

Uh oh!

1996fanrui left a comment

Uh oh!

1996fanrui Oct 15, 2025

Uh oh!

1996fanrui Oct 15, 2025

Uh oh!

1996fanrui Oct 15, 2025

Uh oh!

1996fanrui Oct 15, 2025

Uh oh!

1996fanrui Oct 15, 2025

Uh oh!

1996fanrui Oct 15, 2025

Uh oh!

1996fanrui Oct 15, 2025

Uh oh!

1996fanrui Oct 15, 2025

Uh oh!

1996fanrui Oct 15, 2025

Uh oh!

1996fanrui Oct 16, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

[FLINK-38461] Introduce SinkUpsertMaterializerV2 #27070

Are you sure you want to change the base?

[FLINK-38461] Introduce SinkUpsertMaterializerV2 #27070

Conversation

rkhachatryan commented Sep 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

flinkbot commented Sep 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

1996fanrui left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

rkhachatryan commented Sep 30, 2025 •

edited

Loading

flinkbot commented Sep 30, 2025 •

edited

Loading